package com.sqllog.listenerThread;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.exception.CanalClientException;
import com.sqllog.config.CanalClientConfig;
import com.sqllog.entity.ClientRunnableException;
import com.sqllog.listener.CanalEventListener;
import com.sqllog.messageMutipyThread.ErrorMessageConsumer;
import com.sqllog.messageMutipyThread.MessageConsumer;
import com.sqllog.messageMutipyThread.MessageManagerResource;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.LinkedBlockingQueue;

/**
 *
 * 监听 canal tcp 请求线程
 * @author chennengcheng
 * @date 2020-08-24 17:36
 * @remark
 */

@Slf4j
public class SimpleMessageTransponderRunnable implements MessageTransponder {

    /**
     * custom 连接配置
     */
    private CanalClientConfig serverInstance;

    /**
     * canal 服务指令
     */
    protected final String destination;

    /**
     * canal 客户端的运行状态
     */
    private volatile boolean running = true;

    private List<CanalEventListener> listenerArrays;

    /**
     * 用于记录 消费成功，但是连接被异常中断的消息
     */

    private LinkedBlockingQueue<Long> errorMessageQueue = new LinkedBlockingQueue<>(Integer.MAX_VALUE);

    @Override
    public boolean isRunning() {
        return this.running;
    }

    /**
     * 消费者线程组
     */
    private List<MessageConsumer> messageConsumers = new ArrayList<>();

    @Override
    public boolean stopRunning() {
        messageConsumers.forEach(v -> v.setRunning(false));
        return this.running = false;
    }

    /**
     * 构造方法，初始化参数
     *
     * @param serverInstance    canal 连接配置
     */
    public SimpleMessageTransponderRunnable(CanalClientConfig serverInstance, List<CanalEventListener> listenerArrays) {
        //参数处理
        Objects.requireNonNull(serverInstance, "配置信息不能为空!");
        //参数初始化
        this.destination = serverInstance.getInstanceName();
        this.listenerArrays = listenerArrays;
        this.serverInstance = serverInstance;
    }

    @Override
    public void run() {
        try {
            //若线程正在进行
            log.info("开始连接");
            CanalConnector connector = this.getConnector(this.serverInstance);
            log.info("连接成功");
            boolean connectStatus = true;
            while (this.running) {
                if (connectStatus) {
                    try {
                        this.processOfMessage(connector);
                    }catch (CanalClientException e) {
                        log.error(JSON.toJSONString(e));
                        log.error("连接异常，已断开");
                        connector.disconnect();
                        connectStatus = false;
                        messageConsumers.forEach(v -> v.setRunning(false));
                    }
                }

                if (!this.running) {
                    break;
                }

                log.error("连接" + connector.checkValid());
                log.error("重新连接 " + serverInstance.getHost() + " 实例名: " + serverInstance.getInstanceName());
                try {
                    connector = getConnector(this.serverInstance);
                    log.info("连接成功");
                    connectStatus = true;
                }catch (Exception e) {
                    log.error("连接失败， 准备重试");
                }

                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }catch (Exception e) {
            throw new ClientRunnableException(JSON.toJSONString(e), JSON.toJSONString(this.serverInstance));
        }

        log.info(this.destination + " 监听已经停止");
    }

    private void processOfMessage(CanalConnector connector) throws Exception{
        // 捕获信息的心跳时间
        final int consumerThreadSize = this.serverInstance.getConsumerThreadSize();
        MessageManagerResource resource = new MessageManagerResource(connector, this.serverInstance, destination, errorMessageQueue, listenerArrays);
        ErrorMessageConsumer errorMessageConsumer = new ErrorMessageConsumer(resource);
        new Thread(errorMessageConsumer).start();
        // 先启动消费者，然后再启动生产者，
        for (int i = 0; i < consumerThreadSize; i++) {
            MessageConsumer messageConsumer = new MessageConsumer(resource);
            this.messageConsumers.add(messageConsumer);
            new Thread(messageConsumer).start();//消费者线程
        }

        while (this.running) {
            resource.productMessage();
        }
    }

    /**
     * 获取集群连接地址
     * @param instance
     * @return
     */
    private List<SocketAddress> getClusterConnectorAddress(CanalClientConfig instance){
        //zookeeper 连接集合
        List<SocketAddress> addresses = new ArrayList<>();
        for (String s : instance.getZookeeperAddress()) {
            String[] entry = s.split(":");
            if (entry.length != 2) {
                throw new CanalClientException("zookeeper 地址格式不正确，应该为 ip:port....:" + s);
            }
            //若符合设定规则，先加入集合
            addresses.add(new InetSocketAddress(entry[0], Integer.parseInt(entry[1])));
        }

        return addresses;
    }

    /**
     * 获取单机连接地址
     * @param serverInstance
     * @return
     */
    private InetSocketAddress getSingleConnectorAddress(CanalClientConfig serverInstance){
        return new InetSocketAddress(serverInstance.getHost(), serverInstance.getPort());
    }

    /**
     * 处理 canal 连接实例
     *
     * @param serverInstance
     * @return
     */
    public CanalConnector getConnector(CanalClientConfig serverInstance) {
        log.info("获取连接 " + JSON.toJSONString(serverInstance));
        //声明连接
        CanalConnector connector;
        //是否是集群模式
        if (serverInstance.isClusterEnabled()) {
            List<SocketAddress> clusterConnectorAddress = getClusterConnectorAddress(serverInstance);
            //若集群的话，使用 newClusterConnector 方法初始化
            connector = CanalConnectors.newClusterConnector(clusterConnectorAddress,
                    serverInstance.getInstanceName(), serverInstance.getUserName(), serverInstance.getPassword());
        } else {
            //若不是集群的话，使用 newSingleConnector 初始化
            InetSocketAddress inetSocketAddress = getSingleConnectorAddress(serverInstance);
            connector = CanalConnectors.newSingleConnector(inetSocketAddress,
                    serverInstance.getInstanceName(), serverInstance.getUserName(), serverInstance.getPassword());
        }
        connector.disconnect();
        //canal 连接
        connector.connect();
        if (!StrUtil.isEmpty(serverInstance.getFilter())) {
            //canal 连接订阅，包含过滤规则 订阅数据库表,全部表 connector.subscribe(".*\\..*");
            connector.subscribe(serverInstance.getFilter());
        } else {
            //canal 连接订阅，无过滤规则
            connector.subscribe();
        }

        //canal 回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的地方开始拿
        connector.rollback();
        //返回 canal 连接
        return connector;
    }
}
